package com.liu.network.netty;

import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 服务器的处理逻辑
 */
public class NettyServerHandler1 extends SimpleChannelInboundHandler<Object> {
	public NettyServerHandler1() {
		map.put("tcp:flow", new LinkedBlockingQueue<Object>(500000));
		map.put("udp:flow", new LinkedBlockingQueue<Object>(500000));
		map.put("P:flow:ipv4", new LinkedBlockingQueue<Object>(500000));
		map.put("P:stat:application", new LinkedBlockingQueue<Object>(500000));
		map.put("P:stat:network", new LinkedBlockingQueue<Object>(500000));
		map.put("P:stat:link", new LinkedBlockingQueue<Object>(500000));
		map.put("P:stat:transport", new LinkedBlockingQueue<Object>(500000));
	}

	// 客户端全量数据
	public static ConcurrentHashMap<String, LinkedBlockingQueue<Object>> map = new ConcurrentHashMap<String, LinkedBlockingQueue<Object>>();
	// 打印台数据
	ConcurrentHashMap<String, Integer> consoleMap = null;
	ByteBuf in = null;
	int frameLength = 0;
	// key字节数组-->一条数据的key长度
	int keyLen = 0;
	// value字节数组-->一条数据的value长度
	int valueLen = 0;
	String mapKey = "";
	String mapValue = "";
	ByteBuf keyBuf = null;
	ByteBuf valueBuf = null;
	byte[] keyArray = null;
	byte[] valueArray = null;
	int length = 0;
	long stime = 0L;

	/**
	 * 读取消息通道
	 *
	 * @param context
	 * @param s
	 * @throws Exception
	 */
	protected void channelRead0(ChannelHandlerContext context, Object msg) throws Exception {
		// readByString(context,msg);
		readByBuffer(context, msg);
	}

	/**
	 * 按行读取
	 * 
	 * @param context
	 * @param msg
	 * @throws InterruptedException
	 */
	public void readByString(ChannelHandlerContext context, Object msg) throws InterruptedException {
		String body = (String) msg;

		// 因为map已经初始化定义了需要的key值，而socket客户端传输过来的是全量的数据，所以我们只要取出已经初始化定义的key，并把其他的key值抛弃掉。
		if ("end".equals(body)) {
			System.out.println("客户端：" + (context.channel().remoteAddress()) + " 数据长度：" + length + " 数据大小：" + map.size());
			System.out.println("服务器端耗时：" + (System.currentTimeMillis() - stime) + "ms");
			return;
		}
		mapKey = body.split(";")[0];
		mapValue = body.split(";")[1];
		length++;
		if (map.get(mapKey) != null) {
			map.get(mapKey).put(mapValue);
		}
	}

	/**
	 * 按字节读取
	 * 
	 * @param context
	 * @param msg
	 * @throws Exception
	 */
	public void readByBuffer(ChannelHandlerContext context, Object msg) throws Exception {
		in = (ByteBuf) msg;
		// frameLength = (int) in.getUnsignedInt(4);// 获取key
		frameLength = 287;// 获取头部
		// 当ByteBuf没有达到长度时，return null
		if (in.readableBytes() < frameLength) {
			System.out.println("TCP半包！！！" + in.readableBytes() + "/" + frameLength);
			return;
		}
		// keyLen = (int) in.getUnsignedInt(4);// 获取key
		// valueLen = (int) in.getUnsignedInt(4);// 获取value
		keyLen = 8;// 获取key
		valueLen = 279;// 获取value
		in.skipBytes(12);// 舍弃头部

		int index = in.readerIndex();
		
		keyBuf = in.slice(index, keyLen).retain();// 取出自己定义的packet包返回给ChannelRead
		valueBuf = in.slice(keyLen + index, valueLen).retain();// 取出自己定义的packet包返回给ChannelRead

		keyArray = new byte[keyBuf.capacity()];
		keyBuf.readBytes(keyArray);
		mapKey = new String(keyArray);
		System.out.println(mapKey);

		valueArray = new byte[valueBuf.capacity()];
		valueBuf.readBytes(valueArray);
		mapValue = new String(valueArray);
		System.out.println(mapValue);

		in.readerIndex(0);// 这一步一定要有，不然其实bytebuf的readerIndex没有变，netty会一直从这里开始读取，将readerIndex移动就相当于把前面的数据处理过了废弃掉了。

		length++;
		if (map.get(mapKey) != null) {
			map.get(mapKey).put(mapValue);
		}
	}

	/**
	 * 处理新加的消息通道
	 *
	 * @param ctx
	 * @throws Exception
	 */
	@Override
	public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
	}

	/**
	 * 处理退出消息通道
	 *
	 * @param ctx
	 * @throws Exception
	 */
	@Override
	public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
		ctx.close();
	}

	/**
	 * 在建立连接时发送消息
	 *
	 * @param ctx
	 * @throws Exception
	 */
	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		stime = System.currentTimeMillis();
		// Channel channel = ctx.channel();
		// boolean active = channel.isActive();
		// if (active) {
		// System.out.println("[" + channel.remoteAddress() + "] is online");
		// } else {
		// System.out.println("[" + channel.remoteAddress() + "] is offline");
		// }
		System.out.println("channelActive");
		// ctx.writeAndFlush("[server]: welcome\n");
	}

	/**
	 * 退出时发送消息
	 *
	 * @param ctx
	 * @throws Exception
	 */
	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		Channel channel = ctx.channel();
		if (!channel.isActive()) {
			System.out.println("[" + channel.remoteAddress() + "] is offline");
		} else {
			System.out.println("[" + channel.remoteAddress() + "] is online");
		}
		ctx.close();
	}

	/**
	 * 异常捕获
	 *
	 * @param ctx
	 * @param e
	 * @throws Exception
	 */
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Exception {
		System.out.println("[" + ctx.channel().remoteAddress() + "] throws Exception" + e);
		ctx.close();
		// ctx.close().sync();
	}

	@Override
	protected void messageReceived(ChannelHandlerContext ctx, Object msg) throws Exception {
		// TODO Auto-generated method stub
		
	}

}